package com.ws;


import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.domain.entity.ChatMessage;
import com.exception.BusinessException;
import com.service.ChatMessageService;
import com.util.SyncRedisCacheOffset;
import com.ws.message.ChatMessageDto;
import com.ws.message.ChatMessageVo;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.common.ErrorCode.FORBIDDEN;
import static com.constant.ChatType.PRIVATE;

/**
 * 消息管理器，负责消息的接收与发送，消息的持久化，并写了一些工具类供处理不同类型的消息
 */
@Slf4j
@Component
public class MessageManager {
    public static final long CHAT_TTL = 30;
    private static final TimeUnit CHAT_UNIT = TimeUnit.MINUTES;
    private static final Integer CHAT_THRESHOLD = 10;
    private static final String SYNC_REDIS_CACHE_LOCK = "sync:redis:cache:lock";
    private static final String REDIS_CACHE_OFFSET = "redisCacheOffset:";
    private final StringRedisTemplate stringRedisTemplate;
    private final ChatMessageService chatMessageService;
    private final RedissonClient redissonClient;

    @Autowired
    public MessageManager(RedissonClient redissonClient, StringRedisTemplate stringRedisTemplate, ChatMessageService chatMessageService) {
        this.stringRedisTemplate = stringRedisTemplate;
        this.redissonClient = redissonClient;
        this.chatMessageService = chatMessageService;
    }

    /**
     * 解析消息
     *
     * @param messageFormat
     * @return
     */
    public ChatMessageDto getChatMessageDto(ChatMessageVo messageFormat) {
        ChatMessageDto chatMessageDto = BeanUtil.copyProperties(messageFormat, ChatMessageDto.class);
        chatMessageDto.setSendTime(System.currentTimeMillis());
        return chatMessageDto;
    }

    /**
     * 是否开启私聊会话，获取私聊密钥
     *
     * @return
     */
    //TODO uuid过期时间
    public UUID getPrivateChatKey(Long receiveId, Long senderId) {
        return getUuid(receiveId, senderId, stringRedisTemplate);
    }

    private void syncRedisCache(String key) {
        //当新消息达到阈值后异步去更新数据库
        ZSetOperations<String, String> zSet = stringRedisTemplate.opsForZSet();
        String jsonOffset = stringRedisTemplate.opsForValue().get(REDIS_CACHE_OFFSET + key);
        if (jsonOffset == null || jsonOffset.isEmpty()) {
            return;
        }
        SyncRedisCacheOffset redisCacheOffset = JSONUtil.toBean(jsonOffset, SyncRedisCacheOffset.class);
        long offsetTime = redisCacheOffset.getTime();
        Integer offset = redisCacheOffset.getOffset();
        Long count = zSet.count(key, offsetTime, System.currentTimeMillis());
        if (count != null) {
            if (count >= CHAT_THRESHOLD) {
                RLock lock = redissonClient.getLock(SYNC_REDIS_CACHE_LOCK);
                try {
                    if (lock.tryLock(0, TimeUnit.SECONDS)) {
                        syncRedisCache(key, offsetTime, offset);

                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                } finally {
                    lock.unlock();
                }
            }
        }
    }

    /**
     * 将消息放入redis中，并检查是否达到同步阈值
     *
     * @param chatMessage
     */
    public void addMessageRedisCache(com.domain.entity.ChatMessage chatMessage) {
        ZSetOperations<String, String> zSet = stringRedisTemplate.opsForZSet();
        String key;
        if (chatMessage.getType().equals(PRIVATE)) {
            UUID uuid = getPrivateChatKey(chatMessage.getReceiverId(), chatMessage.getSenderId());
            key = "private:" + uuid;
        } else {
            key = "group:" + chatMessage.getSenderId();
        }
        syncRedisCacheOffset(REDIS_CACHE_OFFSET + key, chatMessage);
        //将聊天数据添加到缓存中
        RLock lock = redissonClient.getLock(SYNC_REDIS_CACHE_LOCK + key);
        try {
            if (lock.tryLock(1, TimeUnit.SECONDS)) {
                zSet.add(key, JSONUtil.toJsonStr(chatMessage), chatMessage.getSendTime());
                stringRedisTemplate.expire(key, CHAT_TTL, CHAT_UNIT);
            }
        } catch (InterruptedException e) {
            log.error("redis锁获取失败", e);
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
        syncRedisCache(key);
    }

    //确定缓存同步的起始值
    private void syncRedisCacheOffset(String key, com.domain.entity.ChatMessage chatMessage) {
        if (Boolean.FALSE.equals(stringRedisTemplate.hasKey(key))) {
            SyncRedisCacheOffset redisCacheOffset = new SyncRedisCacheOffset();
            QueryWrapper<ChatMessage> wrapper = new QueryWrapper<>();
            wrapper.eq("receive_id", chatMessage.getReceiverId())
                    .eq("sender_id", chatMessage.getSenderId())
                    .orderByDesc("send_time")
                    .last("limit 1");
            ChatMessage newMes = chatMessageService.getOne(wrapper);
            if (newMes==null){
                redisCacheOffset.setTime(chatMessage.getSendTime());
            }else {
                redisCacheOffset.setTime(newMes.getSendTime());
            }
            redisCacheOffset.setOffset(0);
            stringRedisTemplate.opsForValue().setIfAbsent(REDIS_CACHE_OFFSET + key, JSONUtil.toJsonStr(redisCacheOffset));
        }
    }

    static UUID getUuid(Long receiveId, Long senderId, StringRedisTemplate stringRedisTemplate) {
        UUID uuid = UUID.randomUUID();
        if (Boolean.TRUE.equals(stringRedisTemplate.opsForValue().setIfAbsent(senderId.toString(), uuid.toString()))) {
            stringRedisTemplate.opsForValue().setIfAbsent(receiveId.toString(), uuid.toString());
        } else {
            uuid = UUID.fromString(Objects.requireNonNull(stringRedisTemplate.opsForValue().get(senderId.toString())));
        }
        return uuid;
    }

    /***
     * 将缓存批量同步到数据库
     * @param key
     * @param offsetTime
     * @param offset
     */
    @Async(value = "threadPool")
    public void syncRedisCache(String key, long offsetTime, Integer offset) {
        log.info("聊天数据达到阈值，开始同步");
        ZSetOperations<String, String> zSet = stringRedisTemplate.opsForZSet();
        Set<String> mesJson = zSet.rangeByScore(key, offsetTime, offset);
        if (CollectionUtil.isEmpty(mesJson)) {
            throw new BusinessException(FORBIDDEN, "消息同步失败");
        }
        //确定下次缓存同步的起始值
        List<com.domain.entity.ChatMessage> messageList = mesJson.stream().map((json) -> {
            return JSONUtil.toBean(json, ChatMessage.class);
        }).collect(Collectors.toList());
        if (!chatMessageService.saveBatch(messageList)) {
            throw new BusinessException(FORBIDDEN, "消息同步失败");
        }
        stringRedisTemplate.delete(REDIS_CACHE_OFFSET + key);
        stringRedisTemplate.delete(key);
    }


}
